package com.ucas.kafka;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.stereotype.Component;

import com.google.protobuf.InvalidProtocolBufferException;
import com.ucas.protobuf.TCPMsgEntity.TcpCmdExecState;
import com.ucas.protobuf.Lns.ReplayResponse;
import com.ucas.util.PersonFactory.Person;

@Component
public class KafkaReceiver {
	@KafkaListener(topics={"vela-tcpmsg"})
	public void velaTcpMsg(ConsumerRecord<String, Object> record) {
        try {
        	TcpCmdExecState p = TcpCmdExecState.parseFrom(record.value().toString().getBytes());
//			System.out.println(p.getSatid());
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}
	@KafkaListener(topics={"engbc"})
	public void engbcMsg(ConsumerRecord<String, Object> record) {
        try {
        	ReplayResponse p = ReplayResponse.parseFrom(record.value().toString().getBytes());
			System.out.println(p.getEor()+"====="+p.getCid()+"===="+p.getEu(0).getVoltage()+"==="+p.getEu(0).getTmid());
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
	}
	@KafkaListener(topics={"zhw-kafka-1"})
	public void processMessage1(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            System.out.println("processMessage-1 get message:" + message);
        }
	}
}
